import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 任务三：优惠券使用时间统计
 * 统计每种优惠券的使用次数和平均消费间隔
 * 输出格式：<Coupon_id> TAB <平均消费间隔>
 */
public class Task3CouponUsageTime {

    // Mapper：统计每个优惠券的使用情况
    public static class CouponMapper extends Mapper<Object, Text, Text, Text> {
        private Text couponId = new Text();
        private Text intervalInfo = new Text();
        private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] fields = value.toString().split(",", -1);
            
            if (fields.length < 7 || fields[0].equals("User_id")) {
                return;
            }

            String coupon = fields[2];        // Coupon_id
            String dateReceived = fields[5];  // Date_received
            String date = fields[6];          // Date

            // 只统计被使用的优惠券（有领取时间和使用时间）
            if (coupon == null || coupon.equals("null") || coupon.isEmpty()) {
                return;
            }
            if (dateReceived == null || dateReceived.equals("null") || dateReceived.isEmpty()) {
                return;
            }
            if (date == null || date.equals("null") || date.isEmpty()) {
                return;
            }

            try {
                Date received = sdf.parse(dateReceived);
                Date used = sdf.parse(date);
                long intervalDays = (used.getTime() - received.getTime()) / (1000 * 60 * 60 * 24);
                
                if (intervalDays >= 0) { // 确保时间间隔有效
                    couponId.set(coupon);
                    intervalInfo.set("used:" + intervalDays);
                    context.write(couponId, intervalInfo);
                }
            } catch (ParseException e) {
                // 忽略日期解析错误
            }
        }
    }

    // Reducer：计算平均消费间隔并统计总使用次数
    public static class CouponReducer extends Reducer<Text, Text, Text, Text> {
        private Text result = new Text();

        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            int count = 0;
            long totalInterval = 0;

            // 统计使用次数和总间隔
            for (Text val : values) {
                String[] parts = val.toString().split(":");
                if (parts.length == 2 && parts[0].equals("used")) {
                    count++;
                    totalInterval += Long.parseLong(parts[1]);
                }
            }

            // 累加到总使用次数计数器
            if (count > 0) {
                context.getCounter("COUPON_STATS", "TOTAL_USAGE_COUNT").increment(count);
                
                double avgInterval = (double) totalInterval / count;
                result.set(count + "\t" + String.format("%.2f", avgInterval));
                context.write(key, result);
            }
        }
    }

    // 第二个Job：根据阈值过滤并排序
    public static class FilterMapper extends Mapper<Object, Text, DoubleWritable, Text> {
        private DoubleWritable interval = new DoubleWritable();
        private Text couponInfo = new Text();
        private long threshold = 0;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            // 从配置中获取阈值
            threshold = context.getConfiguration().getLong("usage.threshold", 0);
            System.out.println("Filter threshold: " + threshold);
        }

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] parts = value.toString().split("\t");
            if (parts.length == 3) {
                try {
                    String couponId = parts[0];
                    int count = Integer.parseInt(parts[1]);
                    double avgInterval = Double.parseDouble(parts[2]);
                    
                    // 过滤：使用次数必须大于阈值（>1%，不是>=1%）
                    if (count > threshold) {
                        // 按平均间隔降序排序（使用负数）
                        interval.set(-avgInterval);
                        couponInfo.set(couponId + "\t" + String.format("%.2f", avgInterval));
                        context.write(interval, couponInfo);
                    }
                } catch (NumberFormatException e) {
                    // 忽略
                }
            }
        }
    }

    public static class FilterReducer extends Reducer<DoubleWritable, Text, Text, Text> {
        private Text emptyKey = new Text("");

        public void reduce(DoubleWritable key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text val : values) {
                context.write(emptyKey, val);
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        
        // Job 1: 计算每个优惠券的使用次数和平均间隔
        Job job1 = Job.getInstance(conf, "Task3 Coupon Usage Time - Step1");
        job1.setJarByClass(Task3CouponUsageTime.class);
        job1.setMapperClass(CouponMapper.class);
        job1.setReducerClass(CouponReducer.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(Text.class);
        
        FileInputFormat.addInputPath(job1, new Path(args[0]));
        Path tempOutput = new Path(args[1] + "_temp");
        FileOutputFormat.setOutputPath(job1, tempOutput);
        
        if (!job1.waitForCompletion(true)) {
            System.exit(1);
        }

        // 获取总使用次数并计算1%阈值
        long totalUsageCount = job1.getCounters()
            .findCounter("COUPON_STATS", "TOTAL_USAGE_COUNT").getValue();
        long threshold = (long) Math.ceil(totalUsageCount * 0.01);
        
        System.out.println("===== Task3 Statistics =====");
        System.out.println("Total usage count: " + totalUsageCount);
        System.out.println("Threshold (>1%): " + threshold);
        System.out.println("============================");

        // Job 2: 根据1%阈值过滤并按平均间隔排序
        Configuration conf2 = new Configuration();
        conf2.setLong("usage.threshold", threshold);
        
        Job job2 = Job.getInstance(conf2, "Task3 Coupon Usage Time - Step2 Filter");
        job2.setJarByClass(Task3CouponUsageTime.class);
        job2.setMapperClass(FilterMapper.class);
        job2.setReducerClass(FilterReducer.class);
        job2.setMapOutputKeyClass(DoubleWritable.class);
        job2.setMapOutputValueClass(Text.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);
        
        FileInputFormat.addInputPath(job2, tempOutput);
        FileOutputFormat.setOutputPath(job2, new Path(args[1]));
        
        System.exit(job2.waitForCompletion(true) ? 0 : 1);
    }
}
